package com.cetc.sdp.kmga.cs.stream;

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;

import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @description:
 * @author： DengQiang
 * @date: 2017/10/16 10:57
 */
public class LazySingletonBroadcast {

    private static Map<Class<? extends Serializable>, Broadcast<? extends Serializable>> broadcastMap = new ConcurrentHashMap<>();

    private static Map<String, Broadcast<? extends Serializable>> broadcastMap2 = new ConcurrentHashMap<>();

    public static <T extends Serializable> Broadcast<T> getOrCreate(JavaSparkContext jsc, Class<T> tClass, T t) {
        return (Broadcast<T>)broadcastMap.computeIfAbsent(tClass, k -> jsc.broadcast(t));
    }

    public static <T extends Serializable> Broadcast<T> getOrCreate(JavaSparkContext jsc, String key, T t) {
        return (Broadcast<T>)broadcastMap2.computeIfAbsent(key, k -> jsc.broadcast(t));
    }

    public static <T extends Serializable> Broadcast<T> get(String key) {
        return (Broadcast<T>)broadcastMap2.get(key);
    }

    public static <T extends Serializable> Broadcast<T> remove(Class<T> tClass) {
        return (Broadcast<T>)broadcastMap.get(tClass);
    }

    public static <T extends Serializable> Broadcast<T> remove(String key) {
        return (Broadcast<T>)broadcastMap2.get(key);
    }
}
